package com.dobby.logger.system.service;

import com.dobby.logger.common.OpenPage;
import com.dobby.logger.system.vo.SystemLog;
import com.dobby.logger.utils.RedisLogUtils;
import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.MultigetSliceQuery;
import me.prettyprint.hector.api.query.QueryResult;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.SolrInputDocument;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * Created by lgq on 2015/5/29.
 */
@Component
public class SystemLogService {

    protected static Cluster cluster;
    protected static Keyspace keyspace;
    protected static String rowKey;
    protected static String columnFamily;
    protected static StringSerializer stringSerializer = StringSerializer.get();

    private static SolrClient client = null;
    private static String SOLR_URL;

    public SystemLogService() {
        Map<String, String> cassandraCfgMap = RedisLogUtils.getCassandraProps();
        cluster = HFactory.getOrCreateCluster(cassandraCfgMap.get("cluster.name"), cassandraCfgMap.get("cluster.hosts"));
        ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();   // 数据库一致性策略
        ccl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);
        keyspace = HFactory.createKeyspace(cassandraCfgMap.get("cassandra.keyspace"), cluster, ccl);  // 创建keyspace
        columnFamily = cassandraCfgMap.get("cassandra.syslogColumnFamily");
    }

    public void addColumn(Object object) {
        SystemLog systemLog  = (SystemLog) object;
        // 1. 先在solr里面添加索引
        addDocsToSolr(systemLog);
        //２.再把数据存储到cassandra中
        Mutator<String> mutator = HFactory.createMutator(keyspace, stringSerializer);
        rowKey = systemLog.getId();
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("id", systemLog.getId()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("companyId", systemLog.getCompanyId()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("userId", systemLog.getUserId()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("ipAddress", systemLog.getIpAddress()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("url", systemLog.getUrl()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("broswer", systemLog.getBroswer()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("operateDate", systemLog.getOperateDate()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("methodName", systemLog.getMethodName()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("params", systemLog.getParams()));
        mutator.addInsertion(rowKey, columnFamily, HFactory.createStringColumn("exceptionMsg", systemLog.getExceptionMsg()));
        mutator.execute();
    }


    public static void addDocsToSolr(SystemLog sysLog) {
        try {
            SolrInputDocument doc = new SolrInputDocument();
            doc.addField("id", sysLog.getId());
            doc.addField("companyId", sysLog.getCompanyId());
            doc.addField("userId", sysLog.getUserId());
            doc.addField("operateDate", sysLog.getOperateDate());
            getSolrClient().add(doc);
            // 更改为使用solr的autoSoftCommit
            getSolrClient().commit();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (SolrServerException e) {
            e.printStackTrace();
        }
    }


    public static SolrClient getSolrClient() {
        Map<String, String> solrCfgMap = RedisLogUtils.getSolrProps();
        SOLR_URL = solrCfgMap.get("solr.url");
        SOLR_URL += solrCfgMap.get("solr.syslogInstance");

        if(client == null) {
            client =  new HttpSolrClient(SOLR_URL);
        }
        return  client;
    }

    public static List<SystemLog> getModelFromCassandra(List<String> rowkeys) {
        List<SystemLog> sysLogBeanList = new ArrayList<SystemLog>();
        if(CollectionUtils.isEmpty(rowkeys)) {
            return sysLogBeanList;
        }
        MultigetSliceQuery<String, String, String> multigetSliceQuery = HFactory.createMultigetSliceQuery(keyspace, stringSerializer, stringSerializer, stringSerializer);
        multigetSliceQuery.setColumnFamily(columnFamily);
        multigetSliceQuery.setKeys(rowkeys);
        multigetSliceQuery.setRange(null, null, false, Integer.MAX_VALUE);
        QueryResult<Rows<String, String, String>> result = multigetSliceQuery.execute();
        Rows<String, String, String> orderedRows = result.get();
        for (Row<String, String, String> r : orderedRows) {
            //  排除错误数据
            if(r.getColumnSlice().getColumns().size() == 0 ) {
                continue;
            }
            if(r.getColumnSlice().getColumns().size() !=12) {
                continue;
            }
            SystemLog syslog = rowTurnToBean(r);
            sysLogBeanList.add(syslog);

        }
        return sysLogBeanList;
    }

    public void deleteColumns(String[] ids) {
        List<String> rowKeys = Arrays.asList(ids);
        // 1. 先把solr中的索引给删除
        deleteDocsFromSolr(rowKeys);
        // 2. 然后删除cassandra中的数据
        Mutator<String> mutator = HFactory.createMutator(keyspace, stringSerializer);
        for(String rowKey : rowKeys) {
            mutator.addDeletion(rowKey, columnFamily, null, stringSerializer);
        }
        mutator.execute();
    }

    /**
     * 删除指定id的index
     * @param ids
     */
    public static void deleteDocsFromSolr(List<String> ids) {
        try {
            getSolrClient().deleteById(ids);
            getSolrClient().commit();
        } catch (SolrServerException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据条件字符串查询
     */
    public static OpenPage<SystemLog> getColumnByCondition(String companyId, String userId, String startDate, String endDate, OpenPage<SystemLog> openPage) {
        // 1. 从solr中得到相应的rowkeys
        List<String> rowKeys = new ArrayList<String>();
        SolrDocumentList list = queryDocsFromSolr(companyId, userId, startDate, endDate, openPage);
        for (int i=0; i<list.size(); i++) {
            String rowkey = (String) list.get(i).getFieldValue("id");
            list.get(i).getFieldValue("id");
            rowKeys.add(rowkey);
        }
        // 2. 使用rowkey在cassandra中查找
        List<SystemLog> sysloglist =  getModelFromCassandra(rowKeys);
        // 3. 设置openPage
        openPage.setRows(sysloglist);
        // 4、 计算出solr中数据索引的总数
//        openPage.setTotal(getCountForSolr());
        openPage.setTotal(10*20); // 数据量太大了，默认显示20页
        return openPage;
    }

    /**
     *  获取solr中索引的总数
     * @return
     */
    public static long getCountForSolr() {
        SolrQuery query = new SolrQuery("*:*");
        query.setStart(0);
        query.setRows(Integer.MAX_VALUE);
        QueryResponse response = null;
        try {
            response = getSolrClient().query(query);
        } catch (SolrServerException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        SolrDocumentList list = response.getResults();
        return list.getNumFound();
    }

    /**
     * 根据条件字符串查询solr
     */
    public static SolrDocumentList queryDocsFromSolr(String companyId, String userId, String startDate, String endDate, OpenPage openPage) {
        String queryString = "";
        // companyId : *company* && userId:*user* && operateDate:[startDate To endDate]
        if(StringUtils.isNotEmpty(companyId)) {
            queryString += "companyId:*"+companyId+"*";
        }
        if(StringUtils.isNotEmpty(userId)) {
            if(StringUtils.isNotEmpty(queryString)) {
                queryString += " && ";
            }
            queryString += " userId:*"+userId+"*";
        }
        startDate = startDate.replace(" ", "T");
        endDate = endDate.replace(" ", "T");
        if(StringUtils.isNotEmpty(startDate) && StringUtils.isNotEmpty(endDate)) {
            if(StringUtils.isNotEmpty(queryString)) {
                queryString += " && ";
            }
            queryString += " operateDate:[" + startDate + " TO " + endDate + "]";
        } else if(StringUtils.isNotEmpty(startDate) && StringUtils.isEmpty(endDate)) {
            if(StringUtils.isNotEmpty(queryString)) {
                queryString += " && ";
            }
            queryString += " operateDate:[" + startDate + " TO *]";
        } else if(StringUtils.isEmpty(startDate) && StringUtils.isNotEmpty(endDate)) {
            if(StringUtils.isNotEmpty(queryString)) {
                queryString += " && ";
            }
            queryString += " operateDate:[* TO " + endDate + "]";
        }
        // 没有设置查询条件
        if(queryString.equals("")) {
            queryString = "*:*";
        }

        SolrQuery query = new SolrQuery(queryString);
        query.setStart(openPage.getFirst()); // 开始下标
        query.setRows(openPage.getPageSize()); // 每页条数
        query.setSort("operateDate", SolrQuery.ORDER.desc);
        QueryResponse response = null;
        try {
            response = getSolrClient().query(query);
        } catch (SolrServerException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        SolrDocumentList list = response.getResults();
        return list;
    }

    /**
     * cassandra中row和bean的转换
     * @param r
     * @return
     */
    public static SystemLog rowTurnToBean(Row<String, String, String> r) {
        SystemLog syslog = new SystemLog();
        syslog.setId(r.getColumnSlice().getColumnByName("id").getValue());
        syslog.setCompanyId(r.getColumnSlice().getColumnByName("companyId").getValue());
        syslog.setUserId(r.getColumnSlice().getColumnByName("userId").getValue());
        syslog.setIpAddress(r.getColumnSlice().getColumnByName("ipAddress").getValue());
        syslog.setUrl(r.getColumnSlice().getColumnByName("url").getValue());
        syslog.setBroswer(r.getColumnSlice().getColumnByName("broswer").getValue());
        syslog.setOperateDate(r.getColumnSlice().getColumnByName("operateDate").getValue());
        syslog.setMethodName(r.getColumnSlice().getColumnByName("methodName").getValue());
        syslog.setParams(r.getColumnSlice().getColumnByName("params").getValue());
        syslog.setExceptionMsg(r.getColumnSlice().getColumnByName("exceptionMsg").getValue());
        return syslog;
    }


    public static void main(String[] args) {
        // 查询出所有的数据
        SolrQuery query = new SolrQuery("*:*");
        query.setRows(Integer.MAX_VALUE);
        QueryResponse response = null;
        try {
            response = SystemLogService.getSolrClient().query(query);
            SolrDocumentList list = response.getResults();
            for (int i=0; i<list.size(); i++) {
                System.out.println(list.get(i));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        //删除solr中的数据
        try {
            SystemLogService.getSolrClient().deleteByQuery("*:*");
            SystemLogService.getSolrClient().commit();
        } catch (SolrServerException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
